package com.jd.rapid.rpc.client;

import com.jd.rapid.rpc.codec.RpcRequest;
import com.jd.rapid.rpc.codec.RpcResponse;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class RpcFuture  implements Future<Object> {

    private RpcRequest request;

    private RpcResponse response;

    private long startTime;

    private static final long TIME_THRESHOLD=5000;

    private List<RpcCallback> pendingCallbacks=new ArrayList<>();

    private Sync sync;

    private ReentrantLock callbackLock=new ReentrantLock();

    /*****用于异步的提交连接请求的线程池************/
    private ThreadPoolExecutor executor=new ThreadPoolExecutor(16,16,
            60, TimeUnit.SECONDS,new ArrayBlockingQueue<>(65536));

    public RpcFuture(RpcRequest request) {
        this.request=request;
        this.startTime=System.currentTimeMillis();
        this.sync=new Sync();
    }

    /**
     *   get 方法是阻塞的，什么时候能拿到结果，只有down方法触发，才能拿到结果
     *   get down 方法之间互斥
     *
     *
     * 实际的回调处理
     * @param rpcResponse
     */
    public void done(RpcResponse rpcResponse) {
        this.response=rpcResponse;
        //释放
        boolean success=sync.release(1);
        if(success){
             //执行回调
             invokeCallBacks();
        }
        //整体rpc调用的耗时时间
        long costTime=System.currentTimeMillis()-startTime;
        if(TIME_THRESHOLD<costTime){
            log.warn("the rpc response time is too slow,requestId:"+request.getRequestId()
            +",costTime:"+costTime);
        }
    }


    /**
     * 依次执行回调函数处理
     */
    private void invokeCallBacks() {
        callbackLock.lock();
        try {
            for (final RpcCallback rpcCallback : pendingCallbacks) {
                runCallBack(rpcCallback);
            }
        }finally {
            callbackLock.unlock();
        }
    }

    private void runCallBack(RpcCallback rpcCallback) {
        final RpcResponse response=this.response;
        executor.submit(new Runnable() {
            @Override
            public void run() {
                    if(response.getThrowable()==null){
                        rpcCallback.success(response.getResult());
                    }else {
                        rpcCallback.failure(response.getThrowable());
                    }
            }
        });
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isDone() {
        return sync.isDone();
    }

    @Override
    public Object get() throws InterruptedException, ExecutionException {
        //阻塞
        sync.acquire(-1);
        if(this.response!=null){
            return this.response.getResult();
        }else {
            return null;
        }
    }

    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        //阻塞
        boolean success=sync.tryAcquireNanos(-1,unit.toNanos(timeout));
        if(success){
            if(this.response!=null){
                return this.response.getResult();
            }else {
                return null;
            }
        }else {
            throw  new RuntimeException("timout excetion requestId:"+this.request.getRequestId()
                    +",class name"+this.request.getClassName()
                    +"method:"+this.request.getMethodName());
        }
    }

    //AQS
    //自己定义的一把锁
    class Sync extends AbstractQueuedSynchronizer{

        private final int done=1;
        private final int pending=0;

        @Override
        protected boolean tryAcquire(int acquires) {
            //获取许可
            return getState()==done?true:false;
        }

        @Override
        protected boolean tryRelease(int releases) {
            if(getState()==pending){
                if(compareAndSetState(pending,done)){
                    return true;
                }
            }
            return false;
        }


        public boolean isDone(){
            return getState()==done;
        }
    }


    /**
     * 可以在应用执行的过程中添加回调处理函数
     * @param callback
     * @return
     */
    public RpcFuture addCallback(RpcCallback callback){
        callbackLock.lock();
        try {
          if(isDone()){
              //处理完，响应
              runCallBack(callback);
          }else{
              this.pendingCallbacks.add(callback);
          }
        }finally {
            callbackLock.unlock();
        }
        return this;
    }
}
